package externalSort;

import java.util.*;
import java.util.concurrent.*;

public class mergeManager {
    private final List<String> runFiles;
    private final String finalOutputFile;
    private final int bufferSize;
    private final int numThreads;
    private final int queueSize;

    public mergeManager(List<String> runFiles, String finalOutputFile, int bufferSize, int numThreads,int queueSize) {
        this.runFiles = new ArrayList<>(runFiles);
        this.finalOutputFile = finalOutputFile;
        this.bufferSize = bufferSize;
        this.numThreads = numThreads;
        this.queueSize=queueSize;
    }

    public String executeMerge() throws InterruptedException {
        if (runFiles.isEmpty()) {
            return null;
        }

        // 使用队列管理待归并的文件
        Queue<String> fileQueue = new LinkedList<>(this.runFiles);
        int mergeRound = 0;

        // 持续归并直到只剩下一个文件
        while (fileQueue.size() > 1) {
            System.out.println("归并轮次 " + mergeRound + ", 剩余文件数: " + fileQueue.size());

            // 创建线程池
            ExecutorService executor = Executors.newFixedThreadPool(numThreads);
            List<Future<MergeResult>> futures = new ArrayList<>();
            List<String> mergedFiles = Collections.synchronizedList(new ArrayList<>());

            // 每次处理一对文件
            while (fileQueue.size() >= 2) {
                String file1 = fileQueue.poll();
                String file2 = fileQueue.poll();

//                String outputFile = "./file/merge/merged_" + mergeRound + "_" + System.currentTimeMillis() + ".txt";
                String outputFile = fileNameGenerator.generateMergeFileName(file1, file2, "./file/merge/");
//                System.out.println("输出文件名称 "+outputFile);

                // 提交归并任务
                MergeTask task = new MergeTask(file1, file2, outputFile, bufferSize,queueSize);
                Future<MergeResult> future = executor.submit(task);
                futures.add(future);
                mergedFiles.add(outputFile);
            }

            // 如果有单个文件剩余，直接加入下一轮
            if (!fileQueue.isEmpty()) {
                mergedFiles.add(fileQueue.poll());
            }

            // 等待所有归并任务完成
            for (Future<MergeResult> future : futures) {
                try {
                    MergeResult result = future.get();
                    if (!result.success()) {
                        System.out.println("归并任务失败: " + result.errorMessage());
                    }
                } catch (ExecutionException e) {
                    System.out.println("归并任务执行异常: " + e.getMessage());
                }
            }

            executor.shutdown();
            executor.awaitTermination(1, TimeUnit.HOURS);

            // 更新文件队列进行下一轮归并
            fileQueue.clear();
            fileQueue.addAll(mergedFiles);
            mergeRound++;
        }

        // 重命名最终文件
        String finalFile = fileQueue.poll();
        if (finalFile != null && !finalFile.equals(finalOutputFile)) {
            java.io.File file = new java.io.File(finalFile);
            file.renameTo(new java.io.File(finalOutputFile));
            finalFile = finalOutputFile;
        }

        System.out.println("归并完成，最终文件: " + finalFile);
        return finalFile;
    }

    // 内部类用于归并任务结果
    private record MergeResult(boolean success, String errorMessage) {}

    // 内部类表示单个归并任务
    private record MergeTask(String file1, String file2, String outputFile,int bufferSize,int queueSize) implements Callable<MergeResult> {
        @Override
        public MergeResult call() {
            try {
                // 创建阻塞队列
                BlockingQueue<List<Integer>> queue1 = new LinkedBlockingQueue<>(queueSize);
                BlockingQueue<List<Integer>> queue2 = new LinkedBlockingQueue<>(queueSize);
                BlockingQueue<List<Integer>> outputQueue = new LinkedBlockingQueue<>(queueSize);

                // 创建生产者和消费者
                mergeProducer producer1 = new mergeProducer(file1, queue1, bufferSize);
                mergeProducer producer2 = new mergeProducer(file2, queue2, bufferSize);
                mergeConsumer consumer = new mergeConsumer(queue1, queue2, outputQueue, outputFile);

//                System.out.println("创建一组归并任务"+file1+" "+file2+" "+outputFile);

                // 创建线程
                Thread thread1 = new Thread(producer1);
                Thread thread2 = new Thread(producer2);
                Thread thread3 = new Thread(consumer);

                // 启动线程
                thread1.start();
                thread2.start();
                thread3.start();

                // 等待完成
                thread1.join();
                thread2.join();
                thread3.join();

                // 删除输入文件
                new java.io.File(file1).delete();
                new java.io.File(file2).delete();

                return new MergeResult(true, null);

            } catch (Exception e) {
                return new MergeResult(false, e.getMessage());
            }
        }
    }
}